package com.study.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/24 21:45
 */
public class TopNExampleTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);

        //1.在创建表的DDL中直接定义时间属性：事件时间
        String createDDL = " CREATE TABLE click_table (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " event_time AS TO_TIMESTAMP( FROM_UNIXTIME( ts / 1000 ) ), " +   //将ts时间戳转换成timestamp类型
                " WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND " +  //设置水位线的时间间隔为1s
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'input/cart.txt', " +    //指定文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";

        tabEnv.executeSql(createDDL);

        //普通top N，选取当前所有用户中浏览量最大的2个
        Table topNResult = tabEnv.sqlQuery(" select user,cnt,row_num " +
                " from ( " +
                "   select *,ROW_NUMBER() OVER(" +
                "           ORDER BY cnt DESC " +
                "       ) as row_num" +
                "    from ( select user,COUNT(url) AS cnt from click_table group by user)" +
                " ) where row_num <= 2 ");

        // tabEnv.toChangelogStream(topNResult).print("top 2：");

        //窗口top N，统计一段时间内的(前2名)活跃用户
        String subQuerySql = " select user,count(url) as cnt,window_start,window_end " +
                " from table(" +
                " TUMBLE( TABLE click_table, DESCRIPTOR(event_time),INTERVAL '10' SECOND )  " +  //滚动窗口
                " ) " +
                " group by user,window_start,window_end";

        Table windTopNResult = tabEnv.sqlQuery(" select user,cnt,row_num,window_start,window_end " +
                " from ( " +
                "   select *,ROW_NUMBER() OVER(" +
                "           PARTITION BY window_start,window_end " +  //根据窗口时间进行分组
                "           ORDER BY cnt DESC " +
                "       ) as row_num" +
                "    from ( " + subQuerySql + ")" +
                " ) where row_num <= 2 ");

        tabEnv.toDataStream(windTopNResult).print("windTopNResult");


        env.execute();
    }
}
